package com.registration;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.data.Stat;

import java.net.InetSocketAddress;
import java.util.Iterator;
import java.util.List;

public class Register {

    public CuratorFramework curatorFramework;

    CacheMap<String,CuratorCache> map;

    public Register(){
        start();
        map = new CacheMap<>(128);
    }

    private CuratorFramework createWithOptions(String connectionString, RetryPolicy retryPolicy,int connectionTimoutMs,int sessionTimeoutMs){
        return CuratorFrameworkFactory
                .builder()
                .connectString(connectionString)
                .retryPolicy(retryPolicy)
                .connectionTimeoutMs(connectionTimoutMs)
                .sessionTimeoutMs(sessionTimeoutMs)
                .build();
    }

    public String retrieveVal(String path) throws Exception {
        byte[] arr = curatorFramework.getData().forPath(path);
        return new String(arr,"UTF-8");
    }

    public List<String> retrieveCatalog(String path) throws Exception {
        List<String> children = curatorFramework.getChildren().forPath(path);
        return children;
    }

    public void deleteNode(String zkPath) throws Exception {
        //启动客户端实例,连接服务器
        curatorFramework.delete().forPath(zkPath);
    }

    public void deleteAllNode(String zkPath) throws Exception{
        curatorFramework.delete().deletingChildrenIfNeeded().forPath(zkPath);
    }

    public void deleteService(String servicePath, InetSocketAddress address) throws Exception{
        String lockPath = "/lock/"+servicePath;
        InterProcessMutex writeLock = new InterProcessReadWriteLock(curatorFramework,lockPath).writeLock();
        writeLock.acquire();
        List<String> strings = retrieveCatalog(servicePath);
        if(strings.size()==1) {
            deleteAllNode(servicePath);
            writeLock.release();
            return;
        }
        String nowPath = servicePath+"/";
        for(int i=strings.size()-1;i>=0;i--){
            String newPath = nowPath+strings.get(i);
            String[] s = retrieveVal(newPath).split(":");
            if(s[0].equals(address.getHostString())){
                deleteNode(newPath);
                break;
            }
        }
        writeLock.release();
    }

    public boolean existNode(String zkPath) throws Exception {
        Stat stat = curatorFramework.checkExists().forPath(zkPath);
        return stat!=null;
    }

    public CuratorCache getWatcher(String workerPath) throws Exception{
        //检查节点是否存在，没有则创建
        if(!map.containsKey(workerPath)){
            synchronized (this){
                if(!map.containsKey(workerPath)){
                    CuratorCache cache = CuratorCache.build(curatorFramework, workerPath);
                    CuratorCacheListener listener = CuratorCacheListener.builder()
                            .forChanges((oldNode, node) -> {
                                if(node.getData()!=null){
                                    System.out.println("old data: " + new String(oldNode.getData()) + " new data:" + new String(node.getData()));
                                }else{
                                    System.out.println("节点删除");
                                }
                            })
                            .forCreates((node) -> {
                                System.out.println("节点被创建：" + node.getPath());
                            })
                            .forDeletes((node) -> {
                                System.out.println("节点被删除：" + node.getPath());
                                if(node.getPath().equals(workerPath)){
                                    System.out.println("根节点删除，退出监控");
                                    map.remove(workerPath);
                                    cache.close();
                                }
                            })
                            .forInitialized(() -> {
                                System.out.println("Cache initialized");
                                map.put(workerPath,cache);
                            })
                            .build();
                    cache.listenable().addListener(listener);
                     cache.start();
                    map.put(workerPath,null);
                }
            }
        }
        CuratorCache cache;
        while((cache =  map.get(workerPath))==null);
        return cache;
    }

    private void start(){
        start(2000,3,"127.0.0.1:2181",2000,2000);
    }

    private void start(int retrySleepTimeMs,int maxRetries,String connction,int connectionTimeoutMs,int sessionTimeoutMs){
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(retrySleepTimeMs,maxRetries);
        curatorFramework = createWithOptions(connction,retryPolicy,connectionTimeoutMs,sessionTimeoutMs);
        curatorFramework.start();
    }

    public synchronized void close(){
        CloseableUtils.closeQuietly(curatorFramework);
        Iterator i = map.values().iterator();
        while(i.hasNext()){
            CuratorCache cache = (CuratorCache) i.next();
            cache.close();
        }
        curatorFramework = null;
    }
}

